package edu.zju.gis.dbfg.model.statistic;

import edu.zju.gis.dbfg.common.exception.ModelFailedException;
import edu.zju.gis.dbfg.model.PyramidConfig;
import edu.zju.gis.dbfg.model.base.Constant;
import edu.zju.gis.dbfg.model.datasource.DataReaderRDD;
import edu.zju.gis.dbfg.model.util.CrsPick;
import edu.zju.gis.dbfg.model.util.GeoItem;
import edu.zju.gis.dbfg.model.util.TileID;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import org.geotools.referencing.CRS;
import org.locationtech.jts.geom.Geometry;
import org.opengis.referencing.crs.CoordinateReferenceSystem;
import org.opengis.referencing.operation.MathTransform;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.*;

/**
 * 跟据输入的extent图层，计算两个LCRA图层下变化的CC的值
 * @author Hu
 */
public class ChangeStatistic {

    private final static Logger logger = LoggerFactory.getLogger(ChangeStatistic.class);

    public static void main (String[] args) throws Exception {
        // Setup environment
        logger.info("Setup Spark Context");
        SparkConf conf = new SparkConf();
        //conf.setMaster("local[4]");
        //conf.setAppName("lalala");
        JavaSparkContext jsc = new JavaSparkContext(conf);

        // Setup model arguments
        logger.info("Setup arguments");
        ChangeStatisticArgs mArgs = ChangeStatisticArgs.initArgs(args);
        if (mArgs == null) {
            String msg = "init args failed, exit";
            logger.error(msg);
            throw new ModelFailedException(ChangeStatistic.class, "main()", msg, args);
        }

        CoordinateReferenceSystem crs1 = CRS.parseWKT(CrsPick.pickByName(mArgs.getCrs1()));
        //CoordinateReferenceSystem crs2 = CRS.parseWKT(mArgs.getCrs2());
        CoordinateReferenceSystem crs3 = CRS.parseWKT(CrsPick.pickByName(mArgs.getCrs3()));
        CoordinateReferenceSystem targetCRS = Constant.DEFAULT_CRS();
        PyramidConfig targetPyramidConfig = Constant.DEFAULT_PYRAMID_CONFIG();

        MathTransform transform1 = null;
        //MathTransform transform2 = null;
        MathTransform transform3 = null;
        if (!crs1.equals(targetCRS)) {
            transform1 = CRS.findMathTransform(crs1, targetCRS);
        }
//        if (!crs2.equals(targetCRS)) {
//            transform2 = CRS.findMathTransform(crs2, targetCRS);
//        }
        if (!crs3.equals(targetCRS)) {
            transform3 = CRS.findMathTransform(crs3, targetCRS);
        }

        // Read data from datasource and map it to geoItem
        String input1 = mArgs.getInput1();
        String type1 = mArgs.getType1();
        Map<String, Object> attrs1 = new HashMap<>();
        attrs1.put("CC1", mArgs.getCcIndex1());  //Fetch the preYear CC field info
        attrs1.put("CC2", mArgs.getCcIndex2());  //Fetch the nextYear CC field info
        attrs1.put("PAC", mArgs.getPacIndex1());
        attrs1.put("AREA", mArgs.getAreaIndex1());
        JavaPairRDD<String,GeoItem> gi1Pair = null;

        String input3 = mArgs.getInput3();
        String type3 = mArgs.getType3();
        Map<String, Object> attrs3 = new HashMap<>();
        JavaRDD<GeoItem> gi3 = DataReaderRDD.readToRDD(type3, jsc, input3, null, attrs3, transform3);
        JavaPairRDD<String,GeoItem> gi3Pair = DataReaderRDD.mapToPair(gi3, targetPyramidConfig);

        if (type1.equals("ES")) {
            gi3Pair.cache();
            List<TileID> tileIDs = gi3Pair.map(t->TileID.fromString(t._1)).distinct().collect();
            gi1Pair = DataReaderRDD.readToPairRDD(type1, jsc, input1, attrs1, tileIDs, targetPyramidConfig);
        } else {
            JavaRDD<GeoItem> gi1 = DataReaderRDD.readToRDD(type1, jsc, input1, null, attrs1, transform1);
            gi1Pair = DataReaderRDD.mapToPair(gi1, targetPyramidConfig);
        }
        // Read data from datasource and map it to geoItem
//        String input2 = mArgs.getInput2();
//        String type2 = mArgs.getType2();
//        Map<String, Object> attrs2 = new HashMap<>();
//        attrs2.put("CC", mArgs.getCcIndex2());
//        attrs2.put("PAC", mArgs.getPacIndex2());
//        attrs2.put("AREA", mArgs.getAreaIndex2());
//        JavaRDD<GeoItem> gi2 = DataReaderRDD.readToRDD(type2, jsc, input2, null, attrs2, transform2);

//        JavaPairRDD<String, GeoItem> gi1Pair = DataReaderRDD.mapToPair(gi1, targetPyramidConfig);
//        JavaPairRDD<String, GeoItem> gi2Pair = DataReaderRDD.mapToPair(gi2, targetPyramidConfig);
//        JavaPairRDD<String, GeoItem> gi3Pair = DataReaderRDD.mapToPair(gi3, targetPyramidConfig);

        JavaPairRDD<String, Tuple2<GeoItem, GeoItem>> joined1 = gi3Pair.join(gi1Pair) ;
        JavaPairRDD<String, GeoItem> changedFeatureRDD = joined1.mapToPair(new PairFunction<Tuple2<String, Tuple2<GeoItem, GeoItem>>, String, GeoItem>() {
            @Override
            public Tuple2<String, GeoItem> call(Tuple2<String, Tuple2<GeoItem, GeoItem>> input) throws Exception {
                GeoItem gi1 = input._2._1(); //extentGeoItem
                GeoItem gi2 = input._2._2(); //input1GeoItem
                Geometry g1 = gi1.getGeometry();
                Geometry g2 = gi2.getGeometry();

                String cc1 = (String)gi2.getProp().get("CC1");
                String cc2 = (String)gi2.getProp().get("CC2");
                //判断 前后CC是否一致
                if(cc1.equals(cc2)) {
                    return new Tuple2<>("EMPTY", null);
                }
                //判断空间相交
                if(!g1.intersects(g2)) {
                    return new Tuple2<>("EMPTY", null);
                }

                Geometry intersection = g1.intersection(g2);
                Double area = intersection.getArea();
                String pac = String.valueOf(gi2.getProp().get("PAC"));
                GeoItem intersectgi= new GeoItem();

                String id = pac+"##"+cc1+"##"+cc2; //需求是按县统计 所以用pac、cc1、cc2作为key做聚合
                intersectgi.setFid(id);
                intersectgi.setGeometry(intersection);

                Map<String, Object> props = new HashMap<>();
                props.put("CC1",cc1);
                props.put("CC2",cc2);
                props.put("PAC",pac);
                props.put("AREA", area);

                intersectgi.setProp(props);

                return new Tuple2<>(id, intersectgi);
            }
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });

        //JavaPairRDD<String, Tuple2<GeoItem, GeoItem>> joined = intersectRDD1.join(gi2Pair);

//        JavaPairRDD<String, GeoItem> changedFeatureRDD = joined.mapToPair(new PairFunction<Tuple2<String, Tuple2<GeoItem, GeoItem>>, String, GeoItem>() {
//            @Override
//            public Tuple2<String, GeoItem> call(Tuple2<String, Tuple2<GeoItem, GeoItem>> input) throws Exception {
//
//                GeoItem gi1 = input._2._1;
//                GeoItem gi2 = input._2._2;
//                Geometry g1 = gi1.getGeometry();
//                Geometry g2 = gi2.getGeometry();
//
//                if (!g1.intersects(g2)) {
//                    return new Tuple2<>("EMPTY", null);
//                }
//
//                String cc1 = (String)gi1.getProp().get("CC");
//                String pac1 = String.valueOf(gi1.getProp().get("PAC"));
//                String cc2 = (String)gi2.getProp().get("CC");
//                String pac2 = String.valueOf(gi2.getProp().get("PAC"));
//
//                if(cc1.equals(cc2) && pac1.equals(pac2)){
//                    return new Tuple2<>("EMPTY", null);
//                }
//
//                GeoItem result = new GeoItem();
//                Geometry intersection = g1.intersection(g2);
//
//                Double area = intersection.getArea();
//
//                String id = gi1.getFid()+"##"+pac1+"##"+pac2+"##"+cc1+"##"+cc2;
//                result.setFid(id);
//                result.setGeometry(intersection);
//                Map<String,Object> props = new HashMap<>();
//                props.put("CC1",cc1);
//                props.put("CC2",cc2);
//                props.put("PAC1",pac1);
//                props.put("PAC2",pac2);
//                props.put("AREA",area);
//                result.setProp(props);
//
//                return new Tuple2<>(id,result);
//            }
//        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
//            @Override
//            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
//                return !input._1.equals("EMPTY");
//            }
//        });
        changedFeatureRDD.cache();
        writeToFile(changedFeatureRDD, "/usr/local/zju/data/test/33/csIntersect.wkt");

        JavaPairRDD<String, GeoItem> resultRDD = changedFeatureRDD.reduceByKey(new Function2<GeoItem, GeoItem, GeoItem>() {
            @Override
            public GeoItem call(GeoItem gi1, GeoItem gi2) throws Exception {
                Double area = (Double)gi1.getProp().get("AREA")+(Double) gi2.getProp().get("AREA");
                GeoItem result = new GeoItem();
                result.setFid(gi1.getFid());
                Map<String, Object> props = new HashMap<>();
                props.put("AREA",area);
                result.setProp(props);
                return result;
            }
        });

        Map<String, GeoItem> result = resultRDD.collectAsMap();

        File file = new File(mArgs.getOutput());
        if (file.exists()) {
            file.delete();
        }
        FileWriter writer = new FileWriter(file);
        for (String key: result.keySet()) {
            String[] fields = key.split("##");
            String cc1 = fields[1];
            String pac1 = fields[0];
            String cc2 = fields[2];
            GeoItem gi = result.get(key);
            Double area = (Double) gi.getProp().get("AREA");
            String s = String.format("%s\t%s\t%s\t%.9f", pac1, cc1, cc2, area);
            writer.write(s+"\n");
        }
        writer.flush();
        writer.close();
    }

    public static void writeToFile(JavaPairRDD<String, GeoItem> rdd, String filename) throws IOException {
        List<Tuple2<String, GeoItem>> featureResult = rdd.collect();
        File featureFile = new File(filename);
        if (featureFile.exists()) featureFile.delete();
        FileWriter featureWriter = new FileWriter(featureFile);
        for(Tuple2<String, GeoItem> feature: featureResult) {
            GeoItem gi = feature._2;
            String id = gi.getFid();
            Geometry geometry = gi.getGeometry();
            String wkt = geometry.toText();
            featureWriter.write(String.format("%s\t%s\n", id, wkt));
        }
        featureWriter.flush();
        featureWriter.close();

    }

}
